package common

import (
	"context"
	"io"
	"shell/utils"
	"sync/atomic"
	"time"

	"github.com/google/uuid"
)

// TimeoutReadwriteCloser 支持超时断开的ReadwriteCloser
type TimeoutReadwriteCloser interface {
	io.ReadWriteCloser
	/*
		如果超时，就调用Close，并返回true
	*/
	CloseIfTimeout() (bool, error)
}

// TimeoutRWCWrap 在原本的ReadwriteCloser上包装一层，以实现超时关闭
type TimeoutRWCWrap struct {
	target         io.ReadWriteCloser
	maxIdleSeconds int64 // 最大空闲时间，为0或负值表示不限制
	lastReadTime   atomic.Int64
	lastWriteTime  atomic.Int64
	hasTimeout     bool // 是否已经超时
}

func NewTimeoutRWCWarp(raw io.ReadWriteCloser, maxIdleSecond int64) *TimeoutRWCWrap {
	if raw == nil {
		return nil
	}
	result := new(TimeoutRWCWrap)
	result.target = raw
	result.maxIdleSeconds = maxIdleSecond
	result.lastReadTime.Store(time.Now().Unix())
	result.lastWriteTime.Store(time.Now().Unix())
	result.hasTimeout = false
	return result
}

func (trwc *TimeoutRWCWrap) CloseIfTimeout() (bool, error) {
	if trwc.maxIdleSeconds <= 0 {
		return false, nil
	}
	if trwc.hasTimeout {
		return true, nil
	}
	now := time.Now().Unix()
	if now-trwc.lastReadTime.Load() > trwc.maxIdleSeconds && now-trwc.lastWriteTime.Load() > trwc.maxIdleSeconds {
		trwc.hasTimeout = true
		return true, trwc.target.Close()
	}
	return false, nil
}

func (trwc *TimeoutRWCWrap) Read(p []byte) (n int, err error) {
	n, err = trwc.target.Read(p)
	if err != nil {
		return n, err
	}
	if trwc.maxIdleSeconds > 0 {
		trwc.lastReadTime.Store(time.Now().Unix())
	}
	return n, nil
}

func (trwc *TimeoutRWCWrap) Write(p []byte) (n int, err error) {
	n, err = trwc.target.Write(p)
	if err != nil {
		return n, err
	}
	if trwc.maxIdleSeconds > 0 {
		trwc.lastWriteTime.Store(time.Now().Unix())
	}
	return n, nil
}

func (trwc *TimeoutRWCWrap) Close() error {
	return trwc.target.Close()
}

func NewTimeoutRWCPairWarp(target, subTarget io.ReadWriteCloser, maxIdleSecond int32, watcher *TimeoutWatcher) *TimeoutRWCPairWrap {
	if target == nil || subTarget == nil || watcher == nil {
		return nil
	}
	result := new(TimeoutRWCPairWrap)
	result.target = target
	result.subTarget = subTarget
	result.maxIdleSeconds = maxIdleSecond
	result.readPoint.Store(0)
	result.writePoint.Store(0)
	result.hasTimeout = false
	result.watcher = watcher
	result.addPoint = int32(watcher.watchInterval.Seconds())
	return result
}

// 超时关闭对，在target超时的时候关闭target和subTarget
type TimeoutRWCPairWrap struct {
	target         io.ReadWriteCloser
	subTarget      io.ReadWriteCloser
	maxIdleSeconds int32 // 最大空闲时间，为0或负值表示不限制
	readPoint      atomic.Int32
	writePoint     atomic.Int32
	hasTimeout     bool // 是否已经超时
	key            string
	watcher        *TimeoutWatcher
	addPoint       int32
}

func (trwc *TimeoutRWCPairWrap) GetSubTarget() io.ReadWriteCloser {
	return trwc.subTarget
}

func (trwc *TimeoutRWCPairWrap) GetTarget() io.ReadWriteCloser {
	return trwc.target
}

// 类似看门狗
func (trwc *TimeoutRWCPairWrap) CloseIfTimeout() (bool, error) {
	if trwc.maxIdleSeconds <= 0 {
		return false, nil
	}
	if trwc.hasTimeout {
		return true, nil
	}
	trwc.readPoint.Add(trwc.addPoint)
	trwc.writePoint.Add(trwc.addPoint)
	if trwc.readPoint.Load() > trwc.maxIdleSeconds && trwc.writePoint.Load() > trwc.maxIdleSeconds {
		trwc.hasTimeout = true
		_ = trwc.subTarget.Close()
		return true, trwc.target.Close()
	}
	return false, nil
}

func (trwc *TimeoutRWCPairWrap) Read(p []byte) (n int, err error) {
	n, err = trwc.target.Read(p)
	if err != nil {
		return n, err
	}
	if trwc.maxIdleSeconds > 0 {
		trwc.readPoint.Store(0)
	}
	return n, nil
}

func (trwc *TimeoutRWCPairWrap) Write(p []byte) (n int, err error) {
	n, err = trwc.target.Write(p)
	if err != nil {
		return n, err
	}
	if trwc.maxIdleSeconds > 0 {
		trwc.writePoint.Store(0)
	}
	return n, nil
}

func (trwc *TimeoutRWCPairWrap) Close() error {
	if trwc.hasTimeout {
		return nil
	}
	trwc.watcher.m.Delete(trwc.key)
	_ = trwc.subTarget.Close()
	return trwc.target.Close()
}

// 开始target到subTarget的转发
func (trwc *TimeoutRWCPairWrap) RedirSync(key string) {
	trwc.key = key
	go func() {
		_, _ = io.Copy(trwc, trwc.subTarget)
	}()
	if trwc.maxIdleSeconds > 0 && !trwc.hasTimeout {
		trwc.watcher.Watch(key, trwc)
	}
	_, _ = io.Copy(trwc.subTarget, trwc)
}

// 开始target到subTarget的转发，且不需要提供一个key
func (trwc *TimeoutRWCPairWrap) RedirSyncNoKey() {
	trwc.key = uuid.NewString()
	go func() { _, _ = io.Copy(trwc, trwc.subTarget) }()
	if trwc.maxIdleSeconds > 0 && !trwc.hasTimeout {
		trwc.watcher.Watch(trwc.key, trwc)
	}
	_, _ = io.Copy(trwc.subTarget, trwc)
}

type TimeoutWatcher struct {
	m             *utils.OrderTimeoutMap
	watchInterval time.Duration
}

// 新建并启动一个超时观察器
func NewTimeoutWatcher(ctxWithCancel context.Context, watchInterval time.Duration) *TimeoutWatcher {
	result := TimeoutWatcher{
		m:             utils.NewOrderTimeoutMap(watchInterval),
		watchInterval: watchInterval,
	}
	go func() {
		ticker := time.NewTicker(watchInterval)
		defer ticker.Stop()
		for {
			select {
			case <-ticker.C:
				result.WatchTimeout()
			case <-ctxWithCancel.Done():
				return
			}
		}
	}()
	return &result
}

// Watch 添加一个监控项
func (tw *TimeoutWatcher) Watch(key string, val TimeoutReadwriteCloser) {
	tw.m.Put(key, val)
}

// WatchTimeout 执行一次超时处理
func (tw *TimeoutWatcher) WatchTimeout() {
	kv := tw.m.CleanTimeout()
	noTimeout := make([]utils.KVPair, 0)
	for _, v := range kv {
		tout, _ := v.V.(TimeoutReadwriteCloser).CloseIfTimeout()
		if !tout {
			noTimeout = append(noTimeout, v)
		}
	}
	if len(noTimeout) > 0 {
		tw.m.PutBatch(noTimeout)
	}
}
